package com.intct.flink.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-14 10:57
 *
 * mysql -->
 */
public class SQLTest09 {
    public static void main(String[] args) {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 1 获取取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 并行度
        env.setParallelism(1);
        // 开启检点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 获取SQL执行环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // mysql-cdc映射表
        tenv.executeSql(
                "create table customers(" +
                        " id INT" +
                        " ,name STRING" +
                        " ,country STRING" +
                        " ,zip STRING" +
                        " ,PRIMARY KEY (id) NOT ENFORCED" +
                        ") with (" +
                        " 'connector' = 'mysql-cdc'" +
                        " ,'hostname' = 'cdh-node'" +
                        " ,'port' = '13306'" +
                        " ,'username' = 'root'" +
                        " ,'password' = 'Test_090110'" +
                        " ,'database-name' = 'm1'" +
                        " ,'table-name' = 'customers'" +
                        ")");

//        tenv.sqlQuery("select * from customers").execute().print();

        // 映射HBase  1 HBase手动创建 create 'intct85:dim_customers', {NAME => 'f1'}
        tenv.executeSql(
                "create table dim_customers(" +
                        " id STRING" +
                        " ,f1 ROW<name STRING" +
                        "     ,country STRING" +
                        "        ,zip STRING>" +
                        " ,PRIMARY KEY (id) NOT ENFORCED" +
                        ") with (" +
                        "    'connector' = 'hbase-2.2'" +
                        "    ,'table-name' = 'intct85:dim_customers'" +
                        "    ,'zookeeper.quorum' = 'cdh-node:2181'" +
                        ")");

        // 插入数据
        tenv.executeSql("insert into dim_customers select CAST(id AS STRING) , row(name, country, zip) as f1 from customers");
    }
}
